Skip to content

Dispose before on_error/on_completed#558

Merged
AlexInLog merged 3 commits into
AlexInLog:v2from
CorentinBT:finish_before_dispose
Apr 11, 2024
Merged

Dispose before on_error/on_completed#558
AlexInLog merged 3 commits into
AlexInLog:v2from
CorentinBT:finish_before_dispose

Conversation

@CorentinBT

@CorentinBT CorentinBT commented Apr 10, 2024

Copy link
Copy Markdown
Contributor

No description provided.

@CorentinBT CorentinBT force-pushed the finish_before_dispose branch from 3023caa to 29c6e42 Compare April 10, 2024 21:35
@CorentinBT CorentinBT marked this pull request as ready for review April 10, 2024 21:35
CorentinBT
CorentinBT commented Apr 10, 2024
View reviewed changes
Comment thread src/rpp/rpp/sources/concat.hpp
if (state->itr.value() == std::cend(state->container))
{
state->observer.on_completed();
state->dispose();

@CorentinBT CorentinBT Apr 10, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

not sure if this one is really needed, same thing if we need a dispose call on line 109

@AlexInLog AlexInLog Apr 11, 2024

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah, actually half of this calls would happen inside on_error/on_completed of downstream

@github-actions

github-actions Bot commented Apr 10, 2024 •
edited
Loading

Copy link
Copy Markdown
Contributor

BENCHMARK RESULTS (AUTOGENERATED)

ci-ubuntu-gcc

General
name rxcpp rpp prev rpp ratio
Subscribe empty callbacks to empty observable 304.04 ns 2.16 ns 2.16 ns 1.00
Subscribe empty callbacks to empty observable via pipe operator 304.34 ns 2.16 ns 2.16 ns 1.00

Sources
name rxcpp rpp prev rpp ratio
from array of 1 - create + subscribe + immediate 692.97 ns 0.31 ns 0.31 ns 1.00
from array of 1 - create + subscribe + current_thread 1032.23 ns 5.25 ns 5.25 ns 1.00
concat_as_source of just(1 immediate) create + subscribe 2245.82 ns 130.66 ns 122.74 ns 1.06
defer from array of 1 - defer + create + subscribe + immediate 733.30 ns 0.31 ns 0.31 ns 1.00
interval - interval + take(3) + subscribe + immediate 2100.16 ns 59.28 ns 59.23 ns 1.00
interval - interval + take(3) + subscribe + current_thread 3068.52 ns 41.92 ns 32.71 ns 1.28

Filtering Operators
name rxcpp rpp prev rpp ratio
immediate_just+take(1)+subscribe 1079.99 ns 0.31 ns 0.31 ns 1.00
immediate_just+filter(true)+subscribe 822.40 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+skip(1)+subscribe 1000.51 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe 868.57 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+first()+subscribe 1237.35 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+last()+subscribe 914.14 ns 0.31 ns 0.31 ns 1.00
immediate_just+take_last(1)+subscribe 1115.24 ns 18.53 ns 17.29 ns 1.07

Schedulers
name rxcpp rpp prev rpp ratio
immediate scheduler create worker + schedule 264.87 ns 2.16 ns 2.16 ns 1.00
current_thread scheduler create worker + schedule 367.52 ns 7.41 ns 6.79 ns 1.09
current_thread scheduler create worker + schedule + recursive schedule 811.80 ns 64.36 ns 63.66 ns 1.01

Transforming Operators
name rxcpp rpp prev rpp ratio
immediate_just+map(v*2)+subscribe 860.10 ns 0.31 ns 0.31 ns 1.00
immediate_just+scan(10, std::plus)+subscribe 893.65 ns 0.31 ns 0.31 ns 1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe 2334.00 ns 174.17 ns 158.20 ns 1.10
immediate_just+buffer(2)+subscribe 1503.46 ns 14.21 ns 13.58 ns 1.05
immediate_just+window(2)+subscribe + subscsribe inner 2343.82 ns 1044.55 ns 1058.72 ns 0.99

Conditional Operators
name rxcpp rpp prev rpp ratio
immediate_just+take_while(false)+subscribe 868.91 ns - - 0.00
immediate_just+take_while(true)+subscribe 817.84 ns 0.31 ns 0.31 ns 1.00

Utility Operators
name rxcpp rpp prev rpp ratio
immediate_just(1)+subscribe_on(immediate)+subscribe 1976.82 ns 0.31 ns 0.31 ns 1.00

Combining Operators
name rxcpp rpp prev rpp ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe 3359.84 ns 181.45 ns 176.84 ns 1.03
immediate_just(1) + merge_with(immediate_just(2)) + subscribe 3637.05 ns 171.01 ns 164.36 ns 1.04
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe - 136.15 ns 132.93 ns 1.02
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe 3494.51 ns 925.48 ns 882.03 ns 1.05
immediate_just(1) + zip(immediate_just(2)) + subscribe 2099.78 ns 220.90 ns 214.26 ns 1.03

Subjects
name rxcpp rpp prev rpp ratio
publish_subject with 1 observer - on_next 34.50 ns 11.72 ns 12.03 ns 0.97

Scenarios
name rxcpp rpp prev rpp ratio
basic sample 1354.50 ns 16.36 ns 15.44 ns 1.06
basic sample with immediate scheduler 1410.76 ns 5.55 ns 5.55 ns 1.00

Aggregating Operators
name rxcpp rpp prev rpp ratio
immediate_just+reduce(10, std::plus)+subscribe 906.56 ns 0.31 ns 0.31 ns 1.00

Error Handling Operators
name rxcpp rpp prev rpp ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe 1027.92 ns 121.61 ns 128.25 ns 0.95

ci-macos

General
name rxcpp rpp prev rpp ratio
Subscribe empty callbacks to empty observable 1418.99 ns 5.53 ns 5.38 ns 1.03
Subscribe empty callbacks to empty observable via pipe operator 1422.33 ns 5.54 ns 5.43 ns 1.02

Sources
name rxcpp rpp prev rpp ratio
from array of 1 - create + subscribe + immediate 2866.07 ns 0.34 ns 0.45 ns 0.75
from array of 1 - create + subscribe + current_thread 4625.21 ns 43.02 ns 11.30 ns 3.81
concat_as_source of just(1 immediate) create + subscribe 10714.13 ns 681.44 ns 405.62 ns 1.68
defer from array of 1 - defer + create + subscribe + immediate 3432.59 ns 0.70 ns 0.28 ns 2.51
interval - interval + take(3) + subscribe + immediate 9406.94 ns 163.78 ns 92.23 ns 1.78
interval - interval + take(3) + subscribe + current_thread 11420.92 ns 156.00 ns 62.79 ns 2.48

Filtering Operators
name rxcpp rpp prev rpp ratio
immediate_just+take(1)+subscribe 4094.84 ns 0.34 ns 0.30 ns 1.14
immediate_just+filter(true)+subscribe 3741.46 ns 0.70 ns 0.29 ns 2.43
immediate_just(1,2)+skip(1)+subscribe 3944.56 ns 0.34 ns 0.28 ns 1.19
immediate_just(1,1,2)+distinct_until_changed()+subscribe 2974.60 ns 0.67 ns 0.75 ns 0.89
immediate_just(1,2)+first()+subscribe 4625.47 ns 0.34 ns 0.30 ns 1.14
immediate_just(1,2)+last()+subscribe 3434.70 ns 0.34 ns 0.28 ns 1.19
immediate_just+take_last(1)+subscribe 6119.01 ns 112.96 ns 75.49 ns 1.50

Schedulers
name rxcpp rpp prev rpp ratio
immediate scheduler create worker + schedule 1586.99 ns 9.08 ns 5.79 ns 1.57
current_thread scheduler create worker + schedule 2227.80 ns 55.28 ns 15.82 ns 3.49
current_thread scheduler create worker + schedule + recursive schedule 4057.64 ns 323.37 ns 177.05 ns 1.83

Transforming Operators
name rxcpp rpp prev rpp ratio
immediate_just+map(v*2)+subscribe 4301.54 ns 0.35 ns 0.29 ns 1.21
immediate_just+scan(10, std::plus)+subscribe 4674.93 ns 0.88 ns 0.60 ns 1.48
immediate_just+flat_map(immediate_just(v*2))+subscribe 7833.56 ns 604.65 ns 476.73 ns 1.27
immediate_just+buffer(2)+subscribe 3602.87 ns 97.00 ns 77.57 ns 1.25
immediate_just+window(2)+subscribe + subscsribe inner 7799.91 ns 3354.51 ns 3000.40 ns 1.12

Conditional Operators
name rxcpp rpp prev rpp ratio
immediate_just+take_while(false)+subscribe 3137.66 ns - - 0.00
immediate_just+take_while(true)+subscribe 3921.98 ns 0.69 ns 0.31 ns 2.22

Utility Operators
name rxcpp rpp prev rpp ratio
immediate_just(1)+subscribe_on(immediate)+subscribe 7100.50 ns 0.34 ns 0.31 ns 1.07

Combining Operators
name rxcpp rpp prev rpp ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe 15248.48 ns 655.21 ns 528.64 ns 1.24
immediate_just(1) + merge_with(immediate_just(2)) + subscribe 12394.28 ns 917.09 ns 534.84 ns 1.71
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe - 688.31 ns 528.51 ns 1.30
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe 11676.15 ns 2756.76 ns 2425.81 ns 1.14
immediate_just(1) + zip(immediate_just(2)) + subscribe 7519.53 ns 1181.79 ns 990.11 ns 1.19

Subjects
name rxcpp rpp prev rpp ratio
publish_subject with 1 observer - on_next 110.10 ns 71.40 ns 63.30 ns 1.13

Scenarios
name rxcpp rpp prev rpp ratio
basic sample 3976.87 ns 147.59 ns 49.58 ns 2.98
basic sample with immediate scheduler 4000.39 ns 21.16 ns 6.37 ns 3.32

Aggregating Operators
name rxcpp rpp prev rpp ratio
immediate_just+reduce(10, std::plus)+subscribe 3422.36 ns 0.34 ns 0.28 ns 1.18

Error Handling Operators
name rxcpp rpp prev rpp ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe 9460.81 ns 5900.99 ns 5356.91 ns 1.10

ci-ubuntu-clang

General
name rxcpp rpp prev rpp ratio
Subscribe empty callbacks to empty observable 274.96 ns 1.55 ns 1.56 ns 0.99
Subscribe empty callbacks to empty observable via pipe operator 273.57 ns 1.54 ns 1.56 ns 0.99

Sources
name rxcpp rpp prev rpp ratio
from array of 1 - create + subscribe + immediate 569.46 ns 0.31 ns 0.31 ns 1.00
from array of 1 - create + subscribe + current_thread 793.00 ns 5.56 ns 5.56 ns 1.00
concat_as_source of just(1 immediate) create + subscribe 1956.47 ns 113.18 ns 112.99 ns 1.00
defer from array of 1 - defer + create + subscribe + immediate 590.01 ns 0.31 ns 0.31 ns 1.00
interval - interval + take(3) + subscribe + immediate 1556.32 ns 58.31 ns 58.31 ns 1.00
interval - interval + take(3) + subscribe + current_thread 2084.15 ns 31.55 ns 31.54 ns 1.00

Filtering Operators
name rxcpp rpp prev rpp ratio
immediate_just+take(1)+subscribe 938.37 ns 0.31 ns 0.31 ns 1.00
immediate_just+filter(true)+subscribe 665.01 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+skip(1)+subscribe 851.27 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe 686.40 ns 0.62 ns 0.31 ns 2.00
immediate_just(1,2)+first()+subscribe 1123.09 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+last()+subscribe 758.63 ns 0.31 ns 0.31 ns 1.00
immediate_just+take_last(1)+subscribe 957.03 ns 0.31 ns 0.31 ns 1.00

Schedulers
name rxcpp rpp prev rpp ratio
immediate scheduler create worker + schedule 206.05 ns 1.54 ns 1.55 ns 0.99
current_thread scheduler create worker + schedule 311.13 ns 5.57 ns 5.57 ns 1.00
current_thread scheduler create worker + schedule + recursive schedule 625.20 ns 58.60 ns 59.82 ns 0.98

Transforming Operators
name rxcpp rpp prev rpp ratio
immediate_just+map(v*2)+subscribe 659.16 ns 0.31 ns 0.31 ns 1.00
immediate_just+scan(10, std::plus)+subscribe 717.08 ns 0.31 ns 0.31 ns 1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe 1815.13 ns 117.08 ns 119.73 ns 0.98
immediate_just+buffer(2)+subscribe 1371.72 ns 13.89 ns 14.82 ns 0.94
immediate_just+window(2)+subscribe + subscsribe inner 2186.31 ns 797.63 ns 804.57 ns 0.99

Conditional Operators
name rxcpp rpp prev rpp ratio
immediate_just+take_while(false)+subscribe 656.96 ns - - 0.00
immediate_just+take_while(true)+subscribe 655.16 ns 0.31 ns 0.31 ns 1.00

Utility Operators
name rxcpp rpp prev rpp ratio
immediate_just(1)+subscribe_on(immediate)+subscribe 1598.25 ns 0.31 ns 0.31 ns 1.00

Combining Operators
name rxcpp rpp prev rpp ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe 2560.83 ns 124.27 ns 126.17 ns 0.98
immediate_just(1) + merge_with(immediate_just(2)) + subscribe 3104.03 ns 122.47 ns 121.24 ns 1.01
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe - 108.58 ns 110.52 ns 0.98
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe 2703.23 ns 730.17 ns 780.24 ns 0.94
immediate_just(1) + zip(immediate_just(2)) + subscribe 1833.69 ns 166.95 ns 168.92 ns 0.99

Subjects
name rxcpp rpp prev rpp ratio
publish_subject with 1 observer - on_next 23.75 ns 13.57 ns 13.88 ns 0.98

Scenarios
name rxcpp rpp prev rpp ratio
basic sample 1039.63 ns 13.89 ns 13.28 ns 1.05
basic sample with immediate scheduler 1023.47 ns 5.86 ns 6.17 ns 0.95

Aggregating Operators
name rxcpp rpp prev rpp ratio
immediate_just+reduce(10, std::plus)+subscribe 741.85 ns 0.31 ns 0.31 ns 1.00

Error Handling Operators
name rxcpp rpp prev rpp ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe 872.26 ns 122.66 ns 124.49 ns 0.99

ci-windows

General
name rxcpp rpp prev rpp ratio
Subscribe empty callbacks to empty observable 581.83 ns 6.72 ns 4.93 ns 1.36
Subscribe empty callbacks to empty observable via pipe operator 598.08 ns 4.93 ns 4.94 ns 1.00

Sources
name rxcpp rpp prev rpp ratio
from array of 1 - create + subscribe + immediate 1176.10 ns 5.55 ns 5.24 ns 1.06
from array of 1 - create + subscribe + current_thread 1445.52 ns 18.82 ns 18.51 ns 1.02
concat_as_source of just(1 immediate) create + subscribe 4639.81 ns 174.16 ns 171.50 ns 1.02
defer from array of 1 - defer + create + subscribe + immediate 1202.57 ns 5.55 ns 5.24 ns 1.06
interval - interval + take(3) + subscribe + immediate 3113.13 ns 134.43 ns 133.62 ns 1.01
interval - interval + take(3) + subscribe + current_thread 3520.69 ns 58.72 ns 58.67 ns 1.00

Filtering Operators
name rxcpp rpp prev rpp ratio
immediate_just+take(1)+subscribe 1870.05 ns 12.87 ns 12.87 ns 1.00
immediate_just+filter(true)+subscribe 1341.91 ns 11.70 ns 11.71 ns 1.00
immediate_just(1,2)+skip(1)+subscribe 2043.05 ns 13.14 ns 13.09 ns 1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe 1385.42 ns 15.78 ns 15.78 ns 1.00
immediate_just(1,2)+first()+subscribe 2114.08 ns 12.95 ns 12.94 ns 1.00
immediate_just(1,2)+last()+subscribe 1495.24 ns 14.11 ns 14.10 ns 1.00
immediate_just+take_last(1)+subscribe 2054.04 ns 60.02 ns 59.48 ns 1.01

Schedulers
name rxcpp rpp prev rpp ratio
immediate scheduler create worker + schedule 492.89 ns 6.17 ns 6.17 ns 1.00
current_thread scheduler create worker + schedule 669.31 ns 18.20 ns 18.20 ns 1.00
current_thread scheduler create worker + schedule + recursive schedule 1109.36 ns 111.44 ns 110.82 ns 1.01

Transforming Operators
name rxcpp rpp prev rpp ratio
immediate_just+map(v*2)+subscribe 1318.50 ns 11.23 ns 11.48 ns 0.98
immediate_just+scan(10, std::plus)+subscribe 1445.67 ns 21.58 ns 21.58 ns 1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe 3596.70 ns 230.62 ns 227.46 ns 1.01
immediate_just+buffer(2)+subscribe 2321.82 ns 59.86 ns 58.91 ns 1.02
immediate_just+window(2)+subscribe + subscsribe inner 4057.82 ns 1549.17 ns 1530.49 ns 1.01

Conditional Operators
name rxcpp rpp prev rpp ratio
immediate_just+take_while(false)+subscribe 1341.46 ns 11.44 ns 11.45 ns 1.00
immediate_just+take_while(true)+subscribe 1338.27 ns 11.70 ns 11.70 ns 1.00

Utility Operators
name rxcpp rpp prev rpp ratio
immediate_just(1)+subscribe_on(immediate)+subscribe 3766.44 ns 7.71 ns 7.40 ns 1.04

Combining Operators
name rxcpp rpp prev rpp ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe 5262.04 ns 247.79 ns 244.12 ns 1.02
immediate_just(1) + merge_with(immediate_just(2)) + subscribe 6486.03 ns 241.16 ns 236.81 ns 1.02
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe - 224.82 ns 236.57 ns 0.95
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe 5859.00 ns 944.64 ns 947.00 ns 1.00
immediate_just(1) + zip(immediate_just(2)) + subscribe 3555.26 ns 535.42 ns 533.86 ns 1.00

Subjects
name rxcpp rpp prev rpp ratio
publish_subject with 1 observer - on_next 37.03 ns 25.91 ns 26.82 ns 0.97

Scenarios
name rxcpp rpp prev rpp ratio
basic sample 2192.90 ns 59.42 ns 59.46 ns 1.00
basic sample with immediate scheduler 2193.39 ns 38.56 ns 37.03 ns 1.04

Aggregating Operators
name rxcpp rpp prev rpp ratio
immediate_just+reduce(10, std::plus)+subscribe 1494.48 ns 19.98 ns 19.99 ns 1.00

Error Handling Operators
name rxcpp rpp prev rpp ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe 1956.78 ns 337.36 ns 339.57 ns 0.99

AlexInLog
AlexInLog reviewed Apr 11, 2024
View reviewed changes
Comment thread src/tests/utils/disposable_observable.hpp
obs.on_error(std::make_exception_ptr(std::runtime_error{""}));
});

op(observable) | rpp::ops::subscribe([](const auto&) {}, [&callback_called](const std::exception_ptr&) { CHECK(!callback_called); });

@AlexInLog AlexInLog Apr 11, 2024 •
edited
Loading

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think we also need to add check over CHECK(callback_called) after subscribe

@codecov

codecov Bot commented Apr 11, 2024

Copy link
Copy Markdown

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 95.44%. Comparing base (721b469) to head (7d09cc2).
Report is 39 commits behind head on v2.

Additional details and impacted files 
@@            Coverage Diff             @@
##               v2     #558      +/-   ##
==========================================
- Coverage   95.53%   95.44%   -0.10%     
==========================================
  Files          80       89       +9     
  Lines        2105     1645     -460     
==========================================
- Hits         2011     1570     -441     
+ Misses         94       75      -19

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

@sonarqubecloud

sonarqubecloud Bot commented Apr 11, 2024

Copy link
Copy Markdown

Quality Gate Passed Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
51.2% Coverage on New Code
0.7% Duplication on New Code

See analysis details on SonarCloud

@AlexInLog AlexInLog merged commit 6fd0680 into AlexInLog:v2 Apr 11, 2024
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@AlexInLog AlexInLog AlexInLog left review comments

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

2 participants

@CorentinBT @AlexInLog